package net.bwie.douyin.dwd.log1.Function;



import net.bwie.douyin.dwd.log1.bean.RoomInteraction;
import net.bwie.douyin.dwd.log1.bean.TikTokBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @ClassName：realTimeInteractionFunction
 * @Author: Vvn
 * @Date: 2025/5/27 15:04
 * @Description: 必须描述类做什么事情, 实现什么功能
 */

public class realTimeInteractionFunction implements
        WindowFunction<TikTokBean, RoomInteraction, String, TimeWindow> {


    private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    @Override
    public void apply(String s, TimeWindow window,
                      Iterable<TikTokBean> input,
                      Collector<RoomInteraction> out) throws Exception {

        String windowStartTime = fastDateFormat.format(window.getStart());
        String windowEndTime = fastDateFormat.format(window.getEnd());


        int interaction = 0;
        int like = 0;
        int share = 0;
        int message = 0;

        for (TikTokBean bean : input) {

            if ("like".equals(bean.getEventType())) {
                like++;
            }
            if ("share".equals(bean.getEventType())) {
                share++;
            }
            if ("message".equals(bean.getEventType())) {
                message++;
            }

            interaction = like + share + message;
            out.collect(new RoomInteraction(windowStartTime, windowEndTime, s, interaction, share, like, message,System.currentTimeMillis()));
        }


    }
}
